package com.biz.primus.ms.search.service;

import com.biz.primus.base.constant.Constant;
import com.biz.primus.base.elasticsearch.entities.ProductEntity;
import com.biz.primus.base.elasticsearch.repository.ProductEsRepository;
import com.biz.primus.base.enums.vendor.VendorType;
import com.biz.primus.base.vo.commodity.ProductIdxIdentityVo;
import com.biz.primus.base.vo.commodity.SaleStatus;
import com.biz.primus.base.vo.search.ProductIdxVo;
import com.biz.primus.common.utils.DateUtil;
import com.biz.primus.common.utils.JsonUtils;
import com.biz.primus.model.depot.vo.DepotRespVo;
import com.biz.primus.ms.search.feign.DepotFeignClient;
import com.biz.primus.ms.search.feign.ProductIdxFeignClient;
import com.biz.primus.ms.search.transform.ProductIdxVo2ProductEntity;
import com.biz.primus.search.stream.ProductSearchIndexChannel;
import com.biz.primus.search.vo.search.ProductIdxReqVo;
import com.biz.primus.search.vo.search.ProductIdxRespVo;
import com.google.common.collect.Lists;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.collections.CollectionUtils;
import org.apache.commons.lang3.StringUtils;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.cloud.stream.annotation.StreamListener;
import org.springframework.messaging.Message;
import org.springframework.stereotype.Service;
import org.springframework.util.StopWatch;

import java.sql.Timestamp;
import java.util.LinkedList;
import java.util.List;
import java.util.Objects;
import java.util.Queue;

/**
 * 商品索引建立
 * @author jiangnan
 * @Description:
 */
@Service
@Slf4j
public class SearchIndexService {

    @Autowired
    private ProductEsRepository productEsRepository;

    @Autowired
    private DepotFeignClient depotService;

    @Autowired
    private ProductIdxFeignClient productIdxService;

    /**
     * 全量索引监听器
     */
    @StreamListener(ProductSearchIndexChannel.PRODUCT_INDEX_ALL)
    public void productIndexAllListener(Message<ProductIdxIdentityVo> message){
        ProductIdxIdentityVo payload = message.getPayload();
        log.info("productIndexAllListener message:{}",payload);
        if (payload.getProductType() != VendorType.TYPE_B.getValue()){
            log.warn("参数错误 不被支持的全量索引类型 {}",payload);
        }
        if (payload.getProductType() == VendorType.TYPE_B.getValue()){
            this.totalIndexOfType(VendorType.TYPE_B);
        }

    }

    @StreamListener(ProductSearchIndexChannel.PRODUCT_INDEX_INCR)
    public void productIndexIncrListener(Message<ProductIdxIdentityVo> message){
        ProductIdxIdentityVo payload = message.getPayload();
        log.info("productIndexIncrListener message:{}",payload);
        if (StringUtils.isBlank(payload.getProductCode())){
            log.info("增量索引未传递商品编码,退出");
            return;
        }
        if (StringUtils.isNotBlank(payload.getDepotCode())){
            //建立单一门店关于该商品的索引
            ProductIdxVo searchIdx = productIdxService.getSearchIdx(payload);
            this.doSave(searchIdx);
        }else {
            //建立所有门店关于该商品的索引
            this.incrOfAllDepot(payload);
        }
        log.info("增量索引结束");
    }



    /**
     * 建立某商品所有门店的索引
     *
     * @param vo
     */
    public void incrOfAllDepot(ProductIdxIdentityVo vo) {
        log.info("请求构建索引商品：{}", vo.getProductCode());
        final Timestamp t1 = DateUtil.now();
        if (Objects.equals(vo.getSaleStatus(), SaleStatus.OFF_SALE)) {
            log.info("delete product code = {}", vo.getProductCode());
            productEsRepository.deleteByProductCode(vo.getProductCode());
            final Timestamp t2 = DateUtil.now();
            log.info("删除索引成功，用时{}", t2.getTime() - t1.getTime());
            return;
        }

        log.debug("updateIncrProductAllIndexDocument ProductIdxIdentityVo: {}", vo);
        final Timestamp t3 = DateUtil.now();
        List<DepotRespVo> allDepotsAndWarehouseDepots = depotService.findAllDepotsFRdb();
        if (CollectionUtils.isNotEmpty(allDepotsAndWarehouseDepots)){
            log.info("findAllDepotsAndWarehouseDepots {}", JsonUtils.obj2Json(allDepotsAndWarehouseDepots));
        }
        final Timestamp t4 = DateUtil.now();
        log.info("获取所有门店，用时{}", t4.getTime() - t3.getTime());
        List<ProductIdxIdentityVo> identityVos = Lists.newArrayList();
        for (DepotRespVo depotRespVo : allDepotsAndWarehouseDepots) {
            ProductIdxIdentityVo idxIdentityVo = new ProductIdxIdentityVo();
            idxIdentityVo.setProductCode(vo.getProductCode());
            idxIdentityVo.setProductType(vo.getProductType());
            idxIdentityVo.setWillBeDeleted(vo.getWillBeDeleted());
            idxIdentityVo.setDepotCode(depotRespVo.getDepotCode());
            idxIdentityVo.setShopId(Long.valueOf(depotRespVo.getId()));
            identityVos.add(idxIdentityVo);
        }
        log.debug("productIdxService getSearchIndices reqVo size [{}]", identityVos.size());
        final Timestamp t5 = DateUtil.now();
        List<ProductIdxVo> searchIndices = productIdxService.getSearchIndices(identityVos);
        final Timestamp t6 = DateUtil.now();
        log.info("构建索引对象成功，用时{}", t6.getTime() - t5.getTime());
        log.debug("resp {}", searchIndices);
        log.info("delete product code = {}", vo.getProductCode());
        final Timestamp t7 = DateUtil.now();
        productEsRepository.deleteByProductCode(vo.getProductCode());
        final Timestamp t8 = DateUtil.now();
        log.info("删除索引成功，用时{}", t8.getTime() - t7.getTime());
        log.info("save product entity size:{}", searchIndices.size());
        final Timestamp t9 = DateUtil.now();
        this.doSave(searchIndices, System.currentTimeMillis());
        final Timestamp t10 = DateUtil.now();
        log.info("新增商品增量索引成功，用时{}", t10.getTime() - t9.getTime());
    }

    /**
     * 根据商品类型建立全量索引
     * @param productType
     */
    public void totalIndexOfType(VendorType productType) {
        if (log.isInfoEnabled()) {
            log.info("totalIndexOfType vo: {}", productType);
        }

        if (productType == null) {
            log.warn("totalIndexOfType request parameter product type is null, request is invalid");
            return;
        }

        if (productType != VendorType.TYPE_B && productType != VendorType.TYPE_A) {
            log.warn("updateTotalProductIndexDocuments request parameter product type is error, product type is: {}", productType);
            return;
        }
        long currentTimeMills = System.currentTimeMillis();
        StopWatch stopWatch = new StopWatch("totalIndexOfType");
        Long totalSize = 0L;
        if (productType == VendorType.TYPE_A) {
            stopWatch.start("typeAIndexAll");
            typeAIndexAll(currentTimeMills);
            stopWatch.stop();
        } else {
            stopWatch.start("typeBIndexAll");
            Boolean aBoolean = productIdxService.tryLockMakeAllTypeBProductIndex();
            log.info("申请B类商品全量索引锁 结果为 {}", aBoolean.toString());
            if (aBoolean) {
                try {
                    totalSize = typeBIndexAll(currentTimeMills);
                } catch (Exception e) {
                    e.printStackTrace();
                } finally {
                    productIdxService.unLockMakeAllTypeBProductIndex();
                    log.info("释放全量索引锁");
                }
            } else {
                log.info("当前锁被占用,跳过执行全量索引");

            }
            stopWatch.stop();
        }
        stopWatch.start("delete unused Index");
        if (totalSize > Constant.TYPE_B_SIZE_LIMIT || productType == VendorType.TYPE_A) {
//            productEsRepository.deleteByProductTypeAndLastUpdateTimeLessThanEqual(vo.getProductType(), currentTimeMills - 1000);
            log.info("取消删除 索引操作");
        } else {
            log.warn("B类全量索引量为 {}  小于120W,判定为异常的全量索引,跳过删除旧索引步骤", totalSize);
        }
        stopWatch.stop();
        log.info(stopWatch.prettyPrint());
    }

    private void typeAIndexAll(long currentTimeMills) {
        ProductIdxReqVo typeAReqVo = new ProductIdxReqVo();
        typeAReqVo.setProductType(VendorType.TYPE_A.getValue());
        // 所有A类商品的索引(当前是直接返回所有A类商品的索引Vo, 如果以后的A类商品变得很多, 可以改造为按照商家构建索引)
        ProductIdxRespVo productIdxRespVo = productIdxService.getProductIndices(typeAReqVo);
        List<ProductIdxVo> idxVos = productIdxRespVo.getIndices();
        this.doSave(idxVos, currentTimeMills);

    }

    private Long typeBIndexAll(long currentTimeMills) {
        log.info("B类商品全量索引");
        List<DepotRespVo> allDepotsAndWarehouseDepots = depotService.findAllDepotsFRdb();
        if (CollectionUtils.isNotEmpty(allDepotsAndWarehouseDepots)){
            log.info("findAllDepotsAndWarehouseDepots {}", JsonUtils.obj2Json(allDepotsAndWarehouseDepots));
        }
        ProductIdxReqVo typeBReqVo = new ProductIdxReqVo();
        typeBReqVo.setProductType(VendorType.TYPE_B.getValue());
        Queue<DepotRespVo> depotQueue = new LinkedList<>(allDepotsAndWarehouseDepots);
        Integer errorCount = 0;
        Long successCount = 0L;
        while (true) {
            DepotRespVo depotRespVo = depotQueue.poll();
            if (depotQueue.size() == 0) {
                break;
            } else {
                if (depotRespVo == null) {
                    log.error("findAllDepotsAndWarehouseDepots 结果中包含null");
                    continue;
                }
                long dt1 = System.currentTimeMillis();
                typeBReqVo.setDepotCode(depotRespVo.getDepotCode());
                typeBReqVo.setDepotId(Long.valueOf(depotRespVo.getId()));
                //使用guava缓存获取商品ro
                typeBReqVo.setUseCatchProduct(Boolean.TRUE);
                ProductIdxRespVo idxRespVo = null;
                try {
                    idxRespVo = productIdxService.getProductIndices(typeBReqVo);
                } catch (Exception e) {
                    e.printStackTrace();
                    errorCount++;
                    log.warn("获取索引失败  失败次数:{}", errorCount);
                    if (errorCount < 300) {
                        depotQueue.offer(depotRespVo);
                    } else {
                        log.error("获取索引失败次数大于 300,放弃建立  {}  的索引", depotRespVo);
                    }
                }
                if (idxRespVo != null) {
                    List<ProductIdxVo> idxVoList = idxRespVo.getIndices();
                    successCount += idxVoList.size();
                    this.doSave(idxVoList, currentTimeMills);
                    long dt2 = System.currentTimeMillis();
                    if (log.isInfoEnabled()) {
                        log.info("depot {} update indices success,productIdx size {}, it cost: {}ms", depotRespVo.getDepotCode(), idxVoList.size(), dt2 - dt1);
                    }
                }
            }
        }
        return successCount;
    }




    public void incrIndex(ProductIdxIdentityVo vo) {
        log.info("调用商品的增量索引 [{}]", vo);
        StopWatch watch = new StopWatch("incrIndex");
        log.debug("productIdxIdentityVo :{}", vo);
        watch.start();
        ProductIdxVo idxVo = productIdxService.getSearchIdx(vo);
        if (log.isDebugEnabled()) {
            log.debug("productIdxVo idxVo: {}", idxVo);
        }
        if (idxVo != null) {
            ProductEntity productEntity = new ProductIdxVo2ProductEntity().apply(idxVo);
            productEntity = productEsRepository.save(productEntity);
            if (log.isDebugEnabled()) {
                log.debug("productEntity after save: {}", productEntity);
            }
        }
        watch.stop();
        log.info(watch.prettyPrint());
    }




    public void delIndexDocuments(ProductIdxIdentityVo vo) {
        if (!vo.getWillBeDeleted()) {
            log.debug("will be deleted is false , return directly");
            return;
        }
        if (StringUtils.isBlank(vo.getProductCode())) {
            log.debug("productcode is blank, return");
            return;
        }

        if (Objects.equals(vo.getProductType(), VendorType.TYPE_A.getValue())) {
            log.debug("product type A");
            productEsRepository.deleteByProductCode(vo.getProductCode());
        } else if (Objects.equals(vo.getProductType(), VendorType.TYPE_B.getValue())) {
            log.debug("product type B");
            productEsRepository.deleteByProductCode(vo.getProductCode());
        }
    }



    /**
     * 保存商品文档记录
     *
     * @param idxVos           商品索引Vo
     * @param currentTimeMills 当前时间毫秒数
     */
    private void doSave(List<ProductIdxVo> idxVos, Long currentTimeMills) {
        List<ProductEntity> entities = Lists.newArrayList();
        ProductIdxVo2ProductEntity transformer = new ProductIdxVo2ProductEntity();
        for (ProductIdxVo idxVo : idxVos) {
            if (idxVo != null) {
                ProductEntity entity = transformer.apply(idxVo);
                assert entity != null;
                entity.setLastUpdateTime(currentTimeMills);
                entities.add(entity);
            }
        }

        if (CollectionUtils.isNotEmpty(entities)) {
            productEsRepository.save(entities);
        }
    }

    private void doSave(ProductIdxVo idxVo){
        ProductEntity entity = new ProductIdxVo2ProductEntity().apply(idxVo);
        productEsRepository.save(entity);
    }
}
